// package com.cw.kafka.starter.configuration;
//
// import java.util.HashMap;
// import java.util.Map;
//
// import org.apache.kafka.clients.consumer.ConsumerConfig;
// import org.apache.kafka.clients.producer.ProducerConfig;
// import org.apache.kafka.common.serialization.StringDeserializer;
// import org.apache.kafka.common.serialization.StringSerializer;
// import org.springframework.beans.factory.annotation.Value;
// import org.springframework.context.annotation.Bean;
// import org.springframework.context.annotation.Configuration;
// import org.springframework.kafka.annotation.EnableKafka;
// import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
// import org.springframework.kafka.config.KafkaListenerContainerFactory;
// import org.springframework.kafka.core.ConsumerFactory;
// import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
// import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
// import org.springframework.kafka.listener.ContainerProperties;
//
// /**
//  * TODO
//  *
//  * @author chenw
//  * @date 2021/1/10
//  */
// @Configuration
// @EnableKafka
// public class KafkaConfiguration {
//     @Value("${bootstrap-servers}")
//     private String bootstrapServers;
//
//     @Bean(name = "simpleFactory")
//     KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
//         ConcurrentKafkaListenerContainerFactory<Integer, String> factory
//             = new ConcurrentKafkaListenerContainerFactory<>();
//         factory.setConsumerFactory(consumerFactory());
//         factory.setConcurrency(3);
//         factory.getContainerProperties()
//             .setPollTimeout(3000);
//         factory.getContainerProperties()
//             .setGroupId("defaultConsumerGroup");
//         //当使用手动提交时必须设置ackMode为MANUAL,否则会报错No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.
//         factory.getContainerProperties()
//             .setAckMode(ContainerProperties.AckMode.MANUAL);
//         factory.getContainerProperties()
//             .setAckCount(10);
//         factory.getContainerProperties()
//             .setAckTime(10000);
//         return factory;
//     }
//
//     //@Bean
//     public ConsumerFactory<Integer, String> consumerFactory() {
//         return new DefaultKafkaConsumerFactory<>(consumerConfigs());
//     }
//
//     //@Bean
//     public Map<String, Object> consumerConfigs() {
//         Map<String, Object> props = new HashMap<>();
//         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//         //键的序列化方式
//         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//         //值的序列化方式
//         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//         return props;
//     }
// }
